package com.dongdongshop.mqconsumer;

import com.alibaba.dubbo.common.json.JSON;
import com.alibaba.fastjson.JSONObject;
import com.dongdongshop.mapper.TbUserScoreAccountMapper;
import com.dongdongshop.pojo.TbOrder;
import com.dongdongshop.pojo.TbUserScoreAccount;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.List;

@RocketMQMessageListener(topic = "addScore",consumerGroup = "consumerXt")
public class MqConsumerListener implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {
    @Autowired
    private TbUserScoreAccountMapper tbUserScoreAccountMapper;
    @Autowired
    private RedisTemplate redisTemplate;
    @Override
    public void onMessage(String s) {//不支持广播消息

    }
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt messageExt = list.get(0);//只发送了一条消息,可以直接get(0)获取
                String msgId = messageExt.getMsgId();//消息ID
                messageExt.getTopic();//消息主题
                int times = messageExt.getReconsumeTimes();
                byte[] body = messageExt.getBody();
                String msg="";
                try {
                    if(!redisTemplate.boundHashOps("maCosumer").hasKey(msgId)) {
                         msg = new String(body);
                        TbUserScoreAccount account = JSONObject.parseObject(msg, TbUserScoreAccount.class);
                        tbUserScoreAccountMapper.insertSelective(account);
                        redisTemplate.boundHashOps("ddShop:mqConsumer").put(msgId,msg);
                    }
                }catch (Exception e){
                    if(times>2){
                        redisTemplate.boundHashOps("ddShop:consumeFail").put(msgId,msg);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return  ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

    }
}
